package edu.xidian.sselab.cloudcourse.inputdata;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class Object_to_HBase {

    private final String tablename = "create_test_table"; //插入数据的表格
    private final String family = "info";  //列族，本项目中只需要一个列族

    public boolean insertRecordsToHBase(List<Record> allRecords) throws IOException { //参数为对象类实例列表

        List<Put> Puts = new ArrayList<>(); //

        for(Record record : allRecords) {
            //1.确定行键 （time##brand##type）
            String rowKey = record.date+"##"+record.brand+"##"+record.type;
            Put put = new Put(Bytes.toBytes(rowKey)); //在hbase数据库中以字节的形式存储
            //2.添加列
            put.addColumn(Bytes.toBytes(family) , Bytes.toBytes("level") , Bytes.toBytes(record.level));
            put.addColumn(Bytes.toBytes(family) , Bytes.toBytes("time") , Bytes.toBytes(record.date));
            put.addColumn(Bytes.toBytes(family) , Bytes.toBytes("country") , Bytes.toBytes(record.nation));
            put.addColumn(Bytes.toBytes(family) , Bytes.toBytes("sales") , Bytes.toBytes(String.valueOf(record.salenumber)));
            put.addColumn(Bytes.toBytes(family) , Bytes.toBytes("price") , Bytes.toBytes(String.valueOf(record.price)));
            Puts.add(put);
        }
        //3.执行数据库插入操作
        HbaseClient hc = new HbaseClient();
        hc.getConnection();
        Table table = hc.getTableByName(tablename);

        try {
            table.put(Puts); //利用Put类插入数据
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }finally {
            table.close(); //关闭表的连接
            hc.closeConnection(); //关闭connection
        }
        return true;
    }
    public static void main(String[] args) throws IOException {
        String filepath = "D:\\test.xlsx"; //excel文件路径

        List<String> family = new ArrayList<>();
        family.add("info"); //列族

        HBaseCreateTable.CreateTable("create_test_table" , family);
        Excel_to_Object eto = new Excel_to_Object();
        Object_to_HBase oth = new Object_to_HBase();
        oth.insertRecordsToHBase(eto.getObject(filepath));

    }
}
